package join;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;

/**
 * Created by zhengjiasen on 2018/3/3.
 */
public class JoinReducer extends Reducer<Text, Text, Text, Text> {


    private static final Text EMPTY_TEXT = new Text("");
    private Text tmp = new Text();
    private ArrayList<Text> listA = new ArrayList<Text>();
    private ArrayList<Text> listB = new ArrayList<Text>();
    private String joinType = null;

    @Override
    public void setup(Context context) {
        joinType = context.getConfiguration().get("join.type");
    }

    @Override
    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        listA.clear();
        listB.clear();
        Iterator<Text> iterator = values.iterator();
        while (iterator.hasNext()) {
            tmp = iterator.next();
            if (tmp.charAt(0) == 'A') {
                listA.add(new Text(tmp.toString().substring(1)));
            } else if (tmp.charAt(0) == 'B') {
                listB.add(new Text(tmp.toString().substring(1)));
            }
        }
        excuteJoinLogic(context);
    }

    private void excuteJoinLogic(Context context) throws IOException, InterruptedException {

        for (Text A : listA) {
            if (!listB.isEmpty()) {
                for (Text B : listB) {
                    context.write(A, B);
                }
            } else {
                context.write(A, EMPTY_TEXT);
            }
        }
    }

}
